package com.klm.easymq.core.impl;

import com.klm.easymq.config.EasyMqProperties;
import com.klm.easymq.core.IdempotentHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

/**
 * Redis幂等处理器实现
 * 使用Redis实现分布式幂等控制
 */
@Slf4j
public class RedisIdempotentHandler implements IdempotentHandler {
    
    private final StringRedisTemplate redisTemplate;
    private final String keyPrefix;
    private final int database;
    
    public RedisIdempotentHandler(EasyMqProperties.Redis redisConfig, StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.keyPrefix = redisConfig.getKeyPrefix();
        this.database = redisConfig.getDatabase();
        log.info("Initialized Redis idempotent handler with keyPrefix={}, database={}", 
                keyPrefix, database);
    }
    
    @Override
    public boolean isProcessed(String key) {
        try {
            String fullKey = keyPrefix + key;
            Boolean exists = redisTemplate.hasKey(fullKey);
            boolean result = exists != null && exists;
            log.debug("Check Redis idempotent key: {}, result: {}", fullKey, result);
            return result;
        } catch (Exception e) {
            log.error("Failed to check Redis idempotent key: {}", key, e);
            return false;
        }
    }
    
    @Override
    public boolean tryAcquire(String key) {
        try {
            String fullKey = keyPrefix + key;
            // 使用SET命令的NX选项，只有当key不存在时才设置
            Boolean result = redisTemplate.opsForValue().setIfAbsent(fullKey, "1");
            boolean acquired = result != null && result;
            log.debug("Try acquire Redis idempotent key: {}, result: {}", fullKey, acquired);
            return acquired;
        } catch (Exception e) {
            log.error("Failed to try acquire Redis idempotent key: {}", key, e);
            return false;
        }
    }
    
    @Override
    public void markProcessed(String key, int expireSeconds) {
        try {
            String fullKey = keyPrefix + key;
            redisTemplate.opsForValue().set(fullKey, "1", expireSeconds, TimeUnit.SECONDS);
            log.debug("Marked Redis idempotent key as processed: {}, expireSeconds: {}", fullKey, expireSeconds);
        } catch (Exception e) {
            log.error("Failed to mark Redis idempotent key as processed: {}", key, e);
        }
    }
    
    @Override
    public void remove(String key) {
        try {
            String fullKey = keyPrefix + key;
            redisTemplate.delete(fullKey);
            log.debug("Removed Redis idempotent key: {}", fullKey);
        } catch (Exception e) {
            log.error("Failed to remove Redis idempotent key: {}", key, e);
        }
    }
    
    @Override
    public String getType() {
        return "redis";
    }
} 